/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package remoting.channel;

import static common.constants.CommonConstants.CONSUMER_SIDE;
import static common.constants.CommonConstants.SIDE_KEY;

import common.constants.CommonConstants;
import common.exception.RemotingException;
import common.model.MagicUrl;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import remoting.store.DataStoreFactory;
import threadpools.ThreadPool;
import threadpools.support.NamedThreadFactory;

public class WrappedChannelHandler implements ChannelHandlerDelegate {

  protected static final Logger logger = LoggerFactory.getLogger(WrappedChannelHandler.class);

  protected static final ExecutorService SHARED_EXECUTOR = Executors
      .newCachedThreadPool(new NamedThreadFactory("MagicSharedHandler", true));

  protected final ExecutorService executor;

  protected final ChannelHandler handler;

  protected final MagicUrl url;

  public WrappedChannelHandler(ChannelHandler handler, MagicUrl url, ThreadPool threadPool) {
    this.handler = handler;
    this.url = url;
    executor = (ExecutorService) threadPool.getExecutor(url);

    String componentKey = CommonConstants.EXECUTOR_SERVICE_COMPONENT_KEY;
    if (CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(SIDE_KEY))) {
      componentKey = CONSUMER_SIDE;
    }
    DataStoreFactory.getInstance().put(componentKey, Integer.toString(url.getPort()), executor);
  }

  public void close() {
    try {
      if (executor != null) {
        executor.shutdown();
      }
    } catch (Throwable t) {
      logger.warn("fail to destroy thread pool of server: " + t.getMessage(), t);
    }
  }

  @Override
  public void connected(Channel channel) throws RemotingException {
    handler.connected(channel);
  }

  @Override
  public void disconnected(Channel channel) throws RemotingException {
    handler.disconnected(channel);
  }

  @Override
  public void sent(Channel channel, Object message) throws RemotingException {
    handler.sent(channel, message);
  }

  @Override
  public void received(Channel channel, Object message) throws RemotingException {
    handler.received(channel, message);
  }

  @Override
  public void caught(Channel channel, Throwable exception) throws RemotingException {
    handler.caught(channel, exception);
  }

  public ExecutorService getExecutor() {
    return executor;
  }

  @Override
  public ChannelHandler getHandler() {
    if (handler instanceof ChannelHandlerDelegate) {
      return ((ChannelHandlerDelegate) handler).getHandler();
    } else {
      return handler;
    }
  }

  public MagicUrl getUrl() {
    return url;
  }

  public ExecutorService getExecutorService() {
    return executor == null || executor.isShutdown() ? SHARED_EXECUTOR : executor;
  }


}
